package cn.itcast.flink.batch;

import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.accumulators.IntCounter;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author lilulu
 */
//批处理中高级特性：案例演示【Flink 中累加器Accumulator使用，统计处理数据集的条目数】
public class BatchAccumulatorDemo {

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    private static class ClickLog {
        //频道ID
        private long channelId;
        //产品的类别ID
        private long categoryId;

        //产品ID
        private long produceId;
        //用户的ID
        private long userId;
        //国家
        private String country;
        //省份
        private String province;
        //城市
        private String city;
        //网络方式
        private String network;
        //来源方式
        private String source;
        //浏览器类型
        private String browserType;
        //进入网站时间
        private Long entryTime;
        //离开网站时间
        private Long leaveTime;
    }

    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        // 2. 数据源-source
        DataStreamSource<String> dataSet = env.readTextFile("datas/click.log");
//        定义累加器  注册累加器  使用累加器进行计数
        SingleOutputStreamOperator<ClickLog> mapDataSet = dataSet.map(
                new RichMapFunction<String, ClickLog>() {
                    private IntCounter counter = new IntCounter();

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        getRuntimeContext().addAccumulator("counter", counter);
                    }

                    public ClickLog map(String value) throws Exception {
                        counter.add(1);
                        return JSON.parseObject(value, ClickLog.class);
                    }
                }
        );
        // 3. 数据转换-transformation
        // 4. 数据终端-sink
        mapDataSet.writeAsText("datas/click.data");
        // 5. 触发执行-execute
        JobExecutionResult batchAccumulatorDemo = env.execute("BatchAccumulatorDemo");
        Object counter = batchAccumulatorDemo.getAccumulatorResult("counter");
        System.out.println("counter:" + counter);
    }
}